package com.xxrl.shop.common.utils;


import com.xxrl.shop.common.entity.ItemSimilarity;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.recommendation.Rating;
import scala.Tuple2;
import scala.Tuple3;
import scala.Tuple4;
import scala.Tuple5;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;

/**
 * @author mis
 * 用来做相似度计算
 *
 * @implNote https://blog.csdn.net/qq_41455420/article/details/89849621
 */
public class SimilarityComputer {

    /**
     * 同现相似度矩阵计算
     * w(i,j) = N(i)nN(j)/sqrt(N(i)*N(j))
     *
     * @param userRdd 用户评分
     * @return 物品相似度
     */
    private JavaRDD<ItemSimilarity> concurrencesSimilarity (JavaRDD<Rating> userRdd) {
        // 1. 数据准备
        JavaRDD<Tuple2<Integer, Integer>> userItemRdd = userRdd.map(x -> Tuple2.apply(x.user(), x.product()));
        JavaPairRDD<Integer, Integer> pairedUserItemRdd = JavaPairRDD.fromJavaRDD(userItemRdd);
        // 2.(用户,物品) 笛卡尔积操作-> 物品与物品组合
        JavaPairRDD<Integer, Tuple2<Integer, Integer>> userItemItemRdd = pairedUserItemRdd.join(pairedUserItemRdd);
        JavaPairRDD<Tuple2<Integer, Integer>, Integer> itemItemRdd = userItemItemRdd.mapToPair(x -> Tuple2.apply(x._2, 1));
        // 3.(物品,物品,频次)
        JavaPairRDD<Tuple2<Integer, Integer>, Integer> itemItemTimeRdd = itemItemRdd.reduceByKey(Integer::sum);
        // 4. 对角矩阵
        JavaPairRDD<Tuple2<Integer, Integer>, Integer> diagonalRdd = itemItemTimeRdd.filter(x -> x._1._1.equals(x._1._2));
        // 5. 非对角矩阵
        JavaPairRDD<Tuple2<Integer, Integer>, Integer> unDiagonalRdd = itemItemTimeRdd.filter(x -> !x._1._1.equals(x._1._2));
        // 6. 计算同现相似度(物品,物品,同现频次
        JavaPairRDD<Integer, Tuple2<Tuple3<Integer, Integer, Integer>, Integer>> rdd8 = unDiagonalRdd
                .mapToPair(x -> Tuple2.apply(x._1._1, Tuple3.apply(x._1._1, x._1._2, x._2)))
                .join(diagonalRdd.mapToPair(x -> Tuple2.apply(x._1._1, x._2)));
        JavaPairRDD<Integer, Tuple4<Integer, Integer, Integer, Integer>> rdd9 = rdd8
                .mapToPair(x -> Tuple2.apply(x._2._1._2(), Tuple4.apply(x._2._1._1(), x._2._1._2(), x._2._1._3(), x._2._2)));

        // 7. 结果返回
        JavaPairRDD<Integer, Tuple2<Tuple4<Integer, Integer, Integer, Integer>, Integer>> rdd10 = rdd9.join(diagonalRdd.mapToPair(x -> Tuple2.apply(x._1._1, x._2)));
        JavaRDD<Tuple5<Integer, Integer, Integer, Integer, Integer>> rdd11 = rdd10.map(x -> Tuple5.apply(x._2._1._1(), x._2._1._2(), x._2._1._3(), x._2._1._4(), x._2._2));
        return rdd11.map(x -> new ItemSimilarity(x._1(), x._2(), (x._3() / Math.sqrt((double)x._4() * x._5()))));
    }

    /**
     * 余弦相似度矩阵计算
     * T(x,y)=∑x(i)y(i)/sqrt(∑(x(i)*y(i))*∑(y(i)*y(i)))
     *
     * @param userRdd 用户评分
     * @return 物品相似度
     */
    private JavaRDD<ItemSimilarity> cosineSimilarity(JavaRDD<Rating> userRdd) {
        // 1. 数据准备
        JavaPairRDD<Integer, Tuple2<Integer, Double>> rdd2 = userRdd.mapToPair(x -> Tuple2.apply(x.user(), Tuple2.apply(x.product(), x.rating())));
        // 2. (用户,物品,评分)笛卡尔积 => (物品1,物品2,评分1,评分2)组合
        JavaPairRDD<Integer, Tuple2<Tuple2<Integer, Double>, Tuple2<Integer, Double>>> rdd3 = rdd2.join(rdd2);
        JavaPairRDD<Tuple2<Integer, Integer>, Tuple2<Double, Double>> rdd4 = rdd3.mapToPair(x -> Tuple2.apply(Tuple2.apply(x._2._1._1, x._2._2._1), Tuple2.apply(x._2._1._2, x._2._2._2)));
        // 3. (物品1, 物品2, 评分1, 评分2)组合 => (物品1,物品2,评分1*评分2)组合并累加
        JavaPairRDD<Tuple2<Integer, Integer>, Double> rdd5 = rdd4.mapToPair(x -> Tuple2.apply(x._1, x._2._1 * x._2._2)).reduceByKey(Double::sum);
        // 4. 对角矩阵
        JavaPairRDD<Tuple2<Integer, Integer>, Double> rdd6 = rdd5.filter(x -> x._1._1.equals(x._1._2));
        // 5. 非对角矩阵
        JavaPairRDD<Tuple2<Integer, Integer>, Double> rdd7 = rdd5.filter(x -> !x._1._1.equals(x._1._2));
        // 6. 计算相似度
        JavaPairRDD<Integer, Tuple2<Tuple3<Integer, Integer, Double>, Double>> rdd8= rdd7.mapToPair(x -> Tuple2.apply(x._1._1, Tuple3.apply(x._1._1, x._1._2, x._2))).join(rdd6.mapToPair(x -> Tuple2.apply(x._1._1, x._2)));
        JavaPairRDD<Integer, Tuple4<Integer, Integer, Double, Double>> rdd9 = rdd8.mapToPair(x -> Tuple2.apply(x._2._1._2(), Tuple4.apply(x._2._1._1(), x._2._1._2(), x._2._1._3(), x._2._2)));
        JavaPairRDD<Integer, Tuple2<Tuple4<Integer, Integer, Double, Double>, Double>> rdd10= rdd9.join(rdd6.mapToPair(x -> Tuple2.apply(x._1._1, x._2)));
        JavaRDD<Tuple5<Integer, Integer, Double, Double, Double>> rdd11 = rdd10.map(x -> Tuple5.apply(x._2._1._1(), x._2._1._2(), x._2._1._3(), x._2._1._4(), x._2._2));
        // 7. 结果返回
        return rdd11.map(x -> new ItemSimilarity(x._1(),x._2(),x._3()/Math.sqrt(x._4()*x._5())));
    }

    /**
     * 计算商品相似度
     * @param userRdd 用户评分
     * @param type 计算相似度的调用类型
     * @return 物品相似度
     */
    public JavaRDD<ItemSimilarity> similarity(JavaRDD<Rating> userRdd, String type) {
        switch (type) {
            case "concurrences":
                return concurrencesSimilarity(userRdd);
            case "cosine":
            default:
                return cosineSimilarity(userRdd);
        }
    }

    public List<List<ItemSimilarity>> train(JavaRDD<Rating> rawData, int num) {
        JavaRDD<ItemSimilarity> similarityRdd = similarity(rawData, "concurrences");
        return recommendForItem(similarityRdd,num);
    }

    /**
     * 用户推荐计算
     * @deprecated
     * @param itemSimilar 物品相似度
     * @param userPrefer   用户评分
     * @param num      推荐数量
     * @return 返回用户推荐物品
     */
    @Deprecated
    public JavaRDD<Rating> recommendForUser(JavaRDD<ItemSimilarity> itemSimilar, JavaRDD<Rating> userPrefer , int num) {
        //1.数据准备
        JavaRDD<Tuple3<Integer, Integer, Double>> rdd1 = itemSimilar.map(x -> Tuple3.apply(x.getProductId1(), x.getProductId2(), x.getSimilar()));
        JavaRDD<Tuple3<Integer, Integer, Double>> userPrefer1 = userPrefer.map(x -> Tuple3.apply(x.user(), x.product(), x.rating()));
        //2.矩阵计算（i行j列join）
        JavaPairRDD<Integer, Tuple2<Tuple2<Integer, Double>, Tuple2<Integer, Double>>> rdd2 = rdd1.mapToPair(x -> Tuple2.apply(x._1(), Tuple2.apply(x._2(), x._3()))).join(userPrefer1.mapToPair(x -> Tuple2.apply(x._2(), Tuple2.apply(x._1(), x._3()))));
        //3.矩阵计算（i行j列相乘）
        JavaPairRDD<Tuple2<Integer, Integer>, Double> rdd3 = rdd2.mapToPair(x -> Tuple2.apply(Tuple2.apply(x._2._2._1, x._2._1._1), x._2._2._2 * x._2._1._2));
        //4.矩阵计算（用户：元素累加求和）
        JavaPairRDD<Tuple2<Integer, Integer>, Double> rdd4 = rdd3.reduceByKey(Double::sum);
        //5.矩阵计算（用户：对结果过滤已有物品）
        JavaPairRDD<Integer, Tuple2<Integer, Double>> rdd5 = rdd4.leftOuterJoin(userPrefer1.mapToPair(x -> Tuple2.apply(Tuple2.apply(x._1(), x._2()), 1))).filter(x -> x._2._2.isPresent()).mapToPair(x -> Tuple2.apply(x._1._1, Tuple2.apply(x._1._2, x._2._1)));
        //6.矩阵计算（用户：用户对结果排序，过滤）
        JavaPairRDD<Integer, Iterable<Tuple2<Integer, Double>>> rdd6 = rdd5.groupByKey();
        JavaRDD<List<Tuple3<Integer,Integer, Double>>> rdd7 = rdd6.map(x -> {
            Iterator<Tuple2<Integer, Double>> iterator = x._2.iterator();
            List<Tuple3<Integer,Integer, Double>> data = new ArrayList<>();
            while (iterator.hasNext()) {
                Tuple2<Integer, Double> next = iterator.next();
                data.add(Tuple3.apply(x._1,next._1,next._2));
            }
            data.sort(Comparator.comparing(Tuple3::_3));
            if (data.size() > num) {
                data = data.subList(0, num);
            }
            return data;
        });

        JavaRDD<Tuple3<Integer, Integer, Double>> rdd8 = rdd7.flatMap(List::iterator);
        return rdd8.map(x -> new Rating(x._1(),x._2(),x._3()));
    }


    /**
     * 商品相似计算
     * @param itemSimilar 物品相似度
     * @param num 最多多少物品
     * @return 每个商品返回一个list,包装了最相似的物品
     */
    private List<List<ItemSimilarity>> recommendForItem(JavaRDD<ItemSimilarity> itemSimilar,int num) {
        JavaPairRDD<Integer, Iterable<ItemSimilarity>> group = itemSimilar.groupBy(ItemSimilarity::getProductId1);
        List<List<ItemSimilarity>> ret = new ArrayList<>();
        for (Tuple2<Integer, Iterable<ItemSimilarity>> integerIterableTuple2 : group.collect()) {
            Iterable<ItemSimilarity> itemSimilarities = integerIterableTuple2._2;
            List<ItemSimilarity> list = new ArrayList<>();
            itemSimilarities.forEach(list::add);
            list.sort(ItemSimilarity::compareTo);
            list.subList(0,Math.min(list.size(),num));
            ret.add(list);
        }
        return ret;
    }
}
